package com.heo.task;

import java.io.IOException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;

import com.heo.task.thread.TaskWaterThared;
import com.heo.task.thread.WaterSendInfo;
import com.heo.task.thread.WaterSendInfoComparator;
import com.heo.util.Utils;

public class SendTaskQueuePriority {

	private static Logger logger = Logger.getLogger(SendTaskQueuePriority.class);

	// 线程池维护线程的最少数量
	private final static int CORE_POOL_SIZE = 5;

	// 线程池维护线程的最大数量
	private final static int MAX_POOL_SIZE = 200;

	// 线程池维护线程所允许的空闲时间
	private final static int KEEP_ALIVE_TIME = 10;

	// 线程池所使用的缓冲队列大小
	private final static int WORK_QUEUE_SIZE = 1000;

	// 如果消息队列没有数据，就休眠，再次获取
	private final static int SLEEP_TIME = 20;

	// 消息队列的最大长度，超过就丢弃
	private final static int MAX_QUEUE_SIZE = 20000;

	private static class SingletonHolder {
		private static final SendTaskQueuePriority INSTANCE = new SendTaskQueuePriority();
	}

	private SendTaskQueuePriority() {
		new Thread(accessPriorityThread).start();
	}

	public static final SendTaskQueuePriority getInstance() {
		return SingletonHolder.INSTANCE;
	}

	public int getReceiveTaskSize() {
		return msgQueue.size();
	}

	// 消息缓冲队列
	 private static PriorityBlockingQueue<WaterSendInfo> msgQueue = new PriorityBlockingQueue<WaterSendInfo>(MAX_QUEUE_SIZE, new  WaterSendInfoComparator());

	final RejectedExecutionHandler handler = new RejectedExecutionHandler() {

		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			WaterSendInfo waterSendInfo = ((TaskWaterThared) r)
					.getWaterSendInfo();
			logger.info(waterSendInfo + "消息放入队列中重新等待执行");
			msgQueue.offer(waterSendInfo);
		}
	};

	final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
			CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS,
			new ArrayBlockingQueue<Runnable>(WORK_QUEUE_SIZE), this.handler);

	// 访问消息缓存的调度线程
	// 查看是否有待定请求，如果有，则创建一个新的CallBackThread，并添加到线程池中
	Runnable accessPriorityThread = new Runnable() {
		@Override
		public void run() {
			try {
				logger.info("处理线程启动成功");
				while (true) {
					if (hasMoreAcquire()) {
						WaterSendInfo waterSendInfo = msgQueue.poll();
						logger.info("获取消息=" + waterSendInfo.toString());
						if (Utils.checkNull(waterSendInfo)) {
							waterSendInfo.setSc(waterSendInfo
									.getSc() + 1);
							Runnable task = new TaskWaterThared(waterSendInfo);
							threadPool.execute(task);
						} else {
							logger.info("消息内容为空丢弃");
						}
					} else {
						Thread.sleep(SLEEP_TIME);
					}
				}
			} catch (InterruptedException e) {
				logger.error("处理线程异常" + e.getMessage());
			} catch (IOException e) {
				logger.error("处理线程异常" + e.getMessage());
			}
		}
	};

	private synchronized boolean hasMoreAcquire() throws IOException {
		boolean bool = !msgQueue.isEmpty();
		if (bool) {
			WaterSendInfo waterSendInfo = msgQueue.peek();
			long sysTime = System.currentTimeMillis();
			if (Utils.checkNull(waterSendInfo)) {
				if (waterSendInfo.getSc() <= waterSendInfo.getSs()) {
					if (waterSendInfo.getSt() <= sysTime) {
						bool = true;
					} else {
						bool = false;
					}

				} else {
					msgQueue.poll();
					logger.info("超过重发次数丢弃" + waterSendInfo.toString());
					bool = false;
				}

			} else {
				msgQueue.poll();
				logger.info("消息内容为空丢弃" + waterSendInfo.toString());
				bool = false;
			}
		}
		return bool;
	}

	public boolean addCallBackMsg(WaterSendInfo waterSendInfo)
			throws IOException {
		if (msgQueue.size() >= MAX_QUEUE_SIZE) {
			logger.info("添加消息失败，执行清空操作。");
			msgQueue.clear();
		}
		boolean bool = msgQueue.offer(waterSendInfo);
		logger.info("添加消息=" + waterSendInfo.toString() + ", 结果=" + bool
				+ ", 当前队列大小=" + getReceiveTaskSize());
		return bool;
	}
}
